package net.bwie.realtime.job;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.function.SourceSplitFunction;
import net.bwie.realtime.jtp.KafkaUtil;
import net.bwie.realtime.jtp.MysqlCdcUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class flinkCdcFromDoris {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 设置检查点
//        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://node101:8020/sm-community-realtime/ckpt");
       //获取数据
        DataStream<String> mysqlStream = MysqlCdcUtil.cdcMysqlEarliest(env, "sca", "cars" ,
                "cars_Log" ,
                "entrance_guard_log" ,
                "ower" ,
                "room");
//                mysqlStream.print();

        // todo 数据转换
        DataStream<String> CarStream = processLog(mysqlStream);
//        CarStream.print();

        env.execute("flinkCdcFromDoris");
    }


    private static DataStream<String> processLog(DataStream<String> mysqlStream) {
        // todo 数据清洗
        DataStream<String> stream = streamClean(mysqlStream);
        // todo 数据分流
        DataStream<String> splitStream = splitStream(stream);
        return splitStream;
    }

    private static DataStream<String> splitStream(DataStream<String> stream) {
        // todo 分流
        OutputTag<String> carInfoLog = new OutputTag<String>("carInfoLog-info") {};
        OutputTag<String> carsLog = new OutputTag<String>("carsLog-info") {};
        OutputTag<String> entranceGuardLog= new OutputTag<String>("entranceGuardLog-info") {};
        OutputTag<String> owerLog = new OutputTag<String>("owerLog-info") {};
        OutputTag<String> roomLog = new OutputTag<String>("roomLog-info") {};

        SingleOutputStreamOperator<String> splitStreamLog = stream.process(
                new SourceSplitFunction(carInfoLog, carsLog, entranceGuardLog, owerLog, roomLog)
        );

        // todo 侧边流输出
        DataStream<String> carsideOutput = splitStreamLog.getSideOutput(carInfoLog);
        DataStream<String> carLogsideOutput = splitStreamLog.getSideOutput(carsLog);
        DataStream<String> entranceguardlogsideOutput = splitStreamLog.getSideOutput(entranceGuardLog);
        DataStream<String> owerOutput = splitStreamLog.getSideOutput(owerLog);
        DataStream<String> roomOutput = splitStreamLog.getSideOutput(roomLog);

        // 将数据写入kafka
//        carsideOutput.print();
        KafkaUtil.producerKafka(carsideOutput,"ods-car-info");

//        owerOutput.print();
        KafkaUtil.producerKafka(owerOutput,"ods-ower-info");
//        roomOutput.print();
        KafkaUtil.producerKafka(roomOutput,"ods-room-info");

//        carLogsideOutput.print();
        KafkaUtil.producerKafka(carLogsideOutput,"ods-cars-Log");
//        entranceguardlogsideOutput.print();
        KafkaUtil.producerKafka(entranceguardlogsideOutput,"ods-entrance-guard-log");

        return splitStreamLog;
    }


    private static DataStream<String> streamClean(DataStream<String> stream) {
        // todo 侧边流输出脏数据
        OutputTag<String> dirtyLog = new OutputTag<String>("dirty-log") {
        };
        SingleOutputStreamOperator<String> cleanStream = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    if (null != jsonObject) {
                        out.collect(value);
                    }
                } catch (Exception e) {
                    ctx.output(dirtyLog, value);
                }

            }
        });
        // c-侧边流输出，找到错误数据
        SideOutputDataStream<String> sideOutput = cleanStream.getSideOutput(dirtyLog);
        KafkaUtil.producerKafka(sideOutput,"ods-car-info-dirty");
        // 正常返回数据流
        return cleanStream;
    }
}
